package spider.zookeeper.monitor;

import org.apache.zookeeper.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import spider.zookeeper.BaseZKClient;
import spider.zookeeper.NodeAction;
import spider.zookeeper.WatcherReactor;
import spider.zookeeper.node.Node;
import spider.zookeeper.util.StringUtils;

import javax.annotation.PostConstruct;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.concurrent.locks.ReentrantReadWriteLock;

/**
 * 集群监控
 *
 * @author Archx[archx@foxmail.com]
 * @date 2017/4/7.
 */
public class ClusterMonitor extends BaseZKClient {

    private static final Logger logger = LoggerFactory.getLogger(ClusterMonitor.class);

    /**
     * 活动的服务器节点
     */
    private List<Node> aliveServers = new ArrayList<>();

    private ReentrantReadWriteLock readWriteLock = new ReentrantReadWriteLock();
    private ReentrantReadWriteLock.WriteLock writeLock = readWriteLock.writeLock();
    private ReentrantReadWriteLock.ReadLock readLock = readWriteLock.readLock();

    private String rootPath; // 根节点路径
    private String busyPath; // 繁忙节点路径
    private boolean webContainer; // 是否web容器部署

    public void setRootPath(String rootPath) {
        this.rootPath = rootPath;
    }

    public void setBusyPath(String busyPath) {
        this.busyPath = busyPath;
    }

    public void setWebContainer(boolean webContainer) {
        this.webContainer = webContainer;
    }

    @PostConstruct
    public void run() {
        // web 容器开启守护线程启动
        if (webContainer) {
            Thread thread = new Thread(new Runnable() {
                @Override
                public void run() {
                    establish();
                }
            });
            thread.setDaemon(true);
            thread.start();
        } else {
            establish();
        }
    }

    @Override
    public void execute() {
        // 创建永久节点
        createNode(rootPath, new byte[0], CreateMode.PERSISTENT);
        createNode(busyPath, new byte[0], CreateMode.PERSISTENT);

        // 观察活动节点的变化
        addWatcher(new NodeAction() {
            @Override
            public void doAction(Watcher watcher) throws InterruptedException, KeeperException {
                zooKeeper.getChildren(rootPath, watcher, new AsyncCallback.ChildrenCallback() {
                    public void processResult(int rc, String path, Object ctx, List<String> children) {
                        switch (KeeperException.Code.get(rc)) {
                            case CONNECTIONLOSS:
                                logger.warn("> Connection loss!");
                                break;
                            case OK:
                                //
                                refresh();
                                break;
                            default:
                                logger.warn("> getChildren failed" + KeeperException
                                        .create(KeeperException.Code.get(rc), path));
                        }
                    }
                }, null);
            }
        }, new WatcherReactor() {
            @Override
            public boolean handle(WatchedEvent watchedEvent) {
                if (watchedEvent.getType().equals(Watcher.Event.EventType.NodeChildrenChanged)) {
                    refresh();
                }
                return true;
            }
        });

        // 观察繁忙节点的变化
        addWatcher(new NodeAction() {
            public void doAction(final Watcher watcher) throws InterruptedException, KeeperException {
                zooKeeper.getChildren(busyPath, watcher, new AsyncCallback.ChildrenCallback() {
                    public void processResult(int rc, String path, Object ctx, List<String> children) {
                        switch (KeeperException.Code.get(rc)) {
                            case CONNECTIONLOSS:
                                logger.warn("> Connection loss!");
                                break;
                            case OK:
                                addBusyFlags();
                                break;
                            default:
                                logger.warn("> getChildren failed" + KeeperException
                                        .create(KeeperException.Code.get(rc), path));
                        }
                    }
                }, null);
            }
        }, new WatcherReactor() {
            public boolean handle(WatchedEvent event) {
                if (event.getType().equals(Watcher.Event.EventType.NodeChildrenChanged)) {
                    writeLock.lock();
                    try {
                        addBusyFlags();
                    } finally {
                        writeLock.unlock();
                    }
                }
                return true;
            }
        });
    }

    /**
     * 选择服务节点
     *
     * @return
     */
    public Node selectServer() {
        readLock.lock();
        try {
            if (!aliveServers.isEmpty()) {
                for (Node node : aliveServers) {
                    if (!node.isBusy()) {
                        return node;
                    }
                }
                // 所有在线服务器都繁忙时随机选择一个服务器
                int index = (int) ((Math.random() * 1000) % aliveServers.size());
                return aliveServers.get(index);
            } else {
                Node node = new Node();
                node.setUnavailable(true);
                return node;
            }
        } finally {
            readLock.unlock();
        }
    }

    /**
     * 刷新节点数据
     */
    private void refresh() {
        writeLock.lock();
        aliveServers.clear();
        try {
            List<String> children = zooKeeper.getChildren(rootPath, false);
            for (String serverNode : children) {
                String nodeData = getNodeData(rootPath + "/" + serverNode);
                if (nodeData != null && !"".equals(nodeData)) {
                    Node node = new Node();
                    node.setNodeId(serverNode);
                    node.setNodeUrl(nodeData);
                    aliveServers.add(node);
                }
            }
            logger.info("> Alive servers: {}", aliveServers);
            addBusyFlags();
        } catch (Exception e) {
            logger.error(StringUtils.printStackTrace(e));
        } finally {
            writeLock.unlock();
        }
    }

    /**
     * 更新节点状态
     */
    private void addBusyFlags() {
        try {
            List<String> children = zooKeeper.getChildren(busyPath, false);
            HashSet<String> busyNodes = new HashSet<String>();
            for (String serverNode : children) {
                busyNodes.add(serverNode);
            }
            logger.info("> Busy servers: {}", busyNodes);
            for (Node aliveServer : aliveServers) {
                aliveServer.setBusy(busyNodes.contains(aliveServer.getNodeId()));
            }
        } catch (Exception e) {
            logger.error(StringUtils.printStackTrace(e));
        }
    }

    public List<Node> getAliveServers() {
        return aliveServers;
    }
}
